package main

import (
	"context"
	"fmt"
	"io/ioutil"

	"github.com/example-user/go-pocket-etl/pkg/core"
	"github.com/example-user/go-pocket-etl/pkg/factory"
	"github.com/example-user/go-pocket-etl/pkg/pipeline"
	"github.com/sirupsen/logrus"
	"github.com/spf13/cobra"
	"gopkg.in/yaml.v2"

	// 匿名导入各组件，用于触发其 init() 函数中的自注册逻辑
	_ "github.com/example-user/go-pocket-etl/pkg/processors/convert_type"
	_ "github.com/example-user/go-pocket-etl/pkg/processors/filter_rows"
	_ "github.com/example-user/go-pocket-etl/pkg/processors/mask_data"
	_ "github.com/example-user/go-pocket-etl/pkg/processors/rename_column"
	_ "github.com/example-user/go-pocket-etl/pkg/processors/select_columns"
	_ "github.com/example-user/go-pocket-etl/pkg/sinks/kafka"
	_ "github.com/example-user/go-pocket-etl/pkg/sinks/mysql"
	_ "github.com/example-user/go-pocket-etl/pkg/sinks/pgsql"
	_ "github.com/example-user/go-pocket-etl/pkg/sinks/stdout"
	_ "github.com/example-user/go-pocket-etl/pkg/sources/csv"
	_ "github.com/example-user/go-pocket-etl/pkg/sources/json"
	_ "github.com/example-user/go-pocket-etl/pkg/sources/mysql"
	_ "github.com/example-user/go-pocket-etl/pkg/sources/pgsql"
)

// version 程序版本号，编译时通过 -ldflags="-X main.version=x.y.z" 注入
var version = "dev"

type Config struct {
	Source struct {
		Type   string                 `yaml:"type"`
		Params map[string]interface{} `yaml:"params"`
	} `yaml:"source"`
	ProcessorsSection struct {
		Processors []pipeline.ProcessorConfig `yaml:"processors"`
	} `yaml:",inline"`
	Sink struct {
		Type   string                 `yaml:"type"`
		Params map[string]interface{} `yaml:"params"`
	} `yaml:"sink"`
	Pipeline pipeline.PipelineConfig `yaml:"pipeline"`
}

var (
	cfgFile string
)

func main() {
	logrus.SetFormatter(&logrus.TextFormatter{
		FullTimestamp:   true,
		TimestampFormat: "2006-01-02 15:04:05",
	})
	logrus.SetLevel(logrus.InfoLevel)

	rootCmd := &cobra.Command{
		Use:   "go-pocket-etl",
		Short: "Go-Pocket-ETL 是一个轻量级、高性能的命令行ETL工具。",
	}

	runCmd := &cobra.Command{
		Use:   "run",
		Short: "运行一个ETL任务",
		Run:   runETL,
	}
	runCmd.Flags().StringVarP(&cfgFile, "config", "c", "", "指定YAML配置文件的路径")
	runCmd.MarkFlagRequired("config")

	validateCmd := &cobra.Command{
		Use:   "validate",
		Short: "校验配置文件的语法和逻辑是否正确",
		Run:   validateConfig,
	}
	validateCmd.Flags().StringVarP(&cfgFile, "config", "c", "", "指定YAML配置文件的路径")
	validateCmd.MarkFlagRequired("config")

	versionCmd := &cobra.Command{
		Use:   "version",
		Short: "显示程序的版本号",
		Run: func(cmd *cobra.Command, args []string) {
			fmt.Printf("Go-Pocket-ETL version %s\n", version)
		},
	}

	rootCmd.AddCommand(runCmd, validateCmd, versionCmd)

	if err := rootCmd.Execute(); err != nil {
		logrus.Fatalf("命令执行出错: %v", err)
	}
}

func parseConfigAndLog() (*Config, error) {
	logrus.Infof("正在读取配置文件: %s", cfgFile)
	configData, err := ioutil.ReadFile(cfgFile)
	if err != nil {
		return nil, fmt.Errorf("读取配置文件 %s 失败: %w", cfgFile, err)
	}

	var cfg Config
	if err := yaml.Unmarshal(configData, &cfg); err != nil {
		return nil, fmt.Errorf("解析配置文件 %s 失败 (YAML语法错误): %w", cfgFile, err)
	}
	logrus.Info("YAML 语法和结构正确")

	logrus.WithFields(logrus.Fields{
		"source_type":      cfg.Source.Type,
		"source_params":    fmt.Sprintf("%v", cfg.Source.Params),
		"processors_count": len(cfg.ProcessorsSection.Processors),
		"sink_type":        cfg.Sink.Type,
		"sink_params":      fmt.Sprintf("%v", cfg.Sink.Params),
		"pipeline_config":  fmt.Sprintf("%+v", cfg.Pipeline),
	}).Info("加载到的配置详情")

	return &cfg, nil
}

func validateConfig(cmd *cobra.Command, args []string) {
	logrus.Info("开始校验配置文件...")

	cfg, err := parseConfigAndLog()
	if err != nil {
		logrus.Fatalf("配置解析失败: %v", err)
	}

	source, err := factory.CreateSource(cfg.Source.Type)
	if err != nil {
		logrus.Fatalf("校验失败: %v", err)
	}
	if err := source.Open(cfg.Source.Params); err != nil {
		logrus.Fatalf("数据源 (Source) 配置校验失败: %v", err)
	}
	defer source.Close()
	logrus.Infof("数据源 (Source) '%s' 配置校验成功", cfg.Source.Type)

	for _, pConfig := range cfg.ProcessorsSection.Processors {
		p, err := factory.CreateProcessor(pConfig.Type)
		if err != nil {
			logrus.Fatalf("校验失败: %v", err)
		}
		if err := p.Open(pConfig.Params); err != nil {
			logrus.Fatalf("处理器 (Processor) '%s' 配置校验失败: %v", pConfig.Type, err)
		}
		defer p.Close()
		logrus.Infof("处理器 (Processor) '%s' 配置校验成功", pConfig.Type)
	}

	sink, err := factory.CreateSink(cfg.Sink.Type)
	if err != nil {
		logrus.Fatalf("校验失败: %v", err)
	}
	if err := sink.Open(cfg.Sink.Params); err != nil {
		logrus.Fatalf("数据汇 (Sink) 配置校验失败: %v", err)
	}
	defer sink.Close()
	logrus.Infof("数据汇 (Sink) '%s' 配置校验成功", cfg.Sink.Type)

	logrus.Info("配置文件校验通过！所有组件及其参数均有效。")
}

func runETL(cmd *cobra.Command, args []string) {
	cfg, err := parseConfigAndLog()
	if err != nil {
		logrus.Fatalf("配置解析失败: %v", err)
	}

	source, err := factory.CreateSource(cfg.Source.Type)
	if err != nil {
		logrus.Fatalf("创建 Source 失败: %v", err)
	}

	processors := make([]core.Processor, 0)
	for _, pConfig := range cfg.ProcessorsSection.Processors {
		p, err := factory.CreateProcessor(pConfig.Type)
		if err != nil {
			logrus.Fatalf("创建 Processor 失败: %v", err)
		}
		processors = append(processors, p)
	}

	sink, err := factory.CreateSink(cfg.Sink.Type)
	if err != nil {
		logrus.Fatalf("创建 Sink 失败: %v", err)
	}

	engine := pipeline.NewEngine(source, processors, sink, cfg.Pipeline)
	logrus.Info("ETL 引擎初始化完毕，准备启动管道...")

	ctx := context.Background()

	if err := engine.Run(ctx, cfg.Source.Params, cfg.ProcessorsSection.Processors, cfg.Sink.Params); err != nil {
		logrus.Errorf("ETL管道执行失败: %v", err)
		return
	}

	logrus.Info("ETL管道任务成功完成。")
}
